package com.wyw.function.learning.demo7;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveTask;
import java.util.stream.LongStream;

/**
 * @Title   分支/合并框架
 * @Description
 * @Author Mr Wu yewen.wu.china@gmail.com
 * Update History:
 * Author        Time            Content
 */
public class DemoClient18 extends RecursiveTask<Long> {

    private Long start;
    private Long end;

    // 临界值
    private long temp = 1000L;

    public DemoClient18(long start, long end) {
        this.start = start;
        this.end = end;
    }

    /**
     * 分支/合并框架的目的是以递归方式将可以并行的任务拆分成更小的任务，然后将每个子任务的结果合并起来生成整体结果。
     * 它是ExecutorService接口的一个实现，他把子任务分配给线程池（ForkJoinPool）中的工作线程
     *
     * 使用RecursiveTask
     * 把任务提交到这个池，必须创建RecursiveTask<R>的一个子类，其中R是并行化任务（以及所有子任务）产生的结果类型，或者如果任务不返回结果，则是RecurisiveAction类型
     *
     * 并行求和
     *
     *      对一个任务调用join方法会阻塞调用方，直到该任务做出结果。所以，有必要在两个子任务的计算都开始之后再调用它。否则，你得到的版本会比原始的顺序算法更慢更复杂，因此每个子任务都必须等待另一个子任务完成才能启动
     *      不应该再RecursiveTask内部使用ForkJoinPool的invoke方法，相反，你应该始终直接调用compute或fork方法，只有顺序代码才应该用invoke来启动并行运算
     *      对子任务调用fork方法可以把它排进ForkJoinPool。同时对左边和右边的子任务调用它似乎很自然，但这样做的效率要比直接对其中一个调用compute低。这样做你可以为其中一个子任务重用同一线程，从而避免在线程池中多分一个任务造成的开销
     *      和并行流一样，并非所有的在多核处理器上使用分支/合并框架就比运算快。
     *
     * 工作窃取
     *      分支/合并框架用工作窃取，每个线程都分配给他的任务保存一个双向链式队列，没完成一个任务，就会从队列头上去出下一个任务开始执行。
     *      某个线程完成后，其他线程很忙，就会随机选一个线程
     */
    public static void main(String[] args) {
        test1();
        test2();
        test3();
    }

    /**
     * 计算方法
     */
    @Override
    protected Long compute() {
        long sum = 0L;

        if((end - start) < temp) {
            for (Long i = start; i <= end; i++) {
                sum += i;
            }
            return sum;
        } else {
            // 分支合并计算
            long middle = (start + end) / 2;
            DemoClient18 forkJoin1 = new DemoClient18(start, middle);

            // 拆分任务    把任务压入线程队列
            forkJoin1.fork();
            DemoClient18 forkJoin2 = new DemoClient18(middle + 1, end);

            //拆分任务    把任务压入线程队列
            forkJoin2.fork();
            return forkJoin1.join() + forkJoin2.join();
        }
    }

    /**
     * 普通
     */
    public static void test1() {
        long start = System.currentTimeMillis();
        Long sum = 0L;

        for (Long i = 1L; i <= 1000000000; i++) {
            sum =+ i;
        }
        long end = System.currentTimeMillis();
        System.out.println("sum = " + sum + " time = " + (end-start) + " ms");
    }

    /**
     * 优化1
     */
    public static void test2() {
        long start = System.currentTimeMillis();

        ForkJoinPool forkJoinPool = new ForkJoinPool();
        DemoClient18 forkJoinDemo = new DemoClient18(1, 1000000000);
        forkJoinPool.submit(forkJoinDemo);

        Long sum = 0L;
        try {
            sum = forkJoinDemo.get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }

        long end = System.currentTimeMillis();
        System.out.println("sum = " + sum + " time = " + (end-start) + " ms");
    }

    /**
     * 优化2
     */
    public static void test3() {
        long start = System.currentTimeMillis();

        // Stream流
        long sum = LongStream.rangeClosed(0L, 1000000000L).parallel().reduce(0, Long::sum);

        long end = System.currentTimeMillis();
        System.out.println("sum = " + sum + " time = " + (end-start) + " ms");
    }


}
